package com.wtw.graph

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.graphx.{Edge, EdgeContext, Graph, VertexId}

//只要某个顶点收到了两条msg就会进行merge函数 之后每收到一条就会与之前两条msg运算后的结果进行运算

object aggregateMessageDemo2 {

  var flag = false

  def sumEdgeCount(g: Graph[(Int, Int), String]): Graph[(Int, Int), String] = {
    val verts = g.aggregateMessages[(Int, Int)](send, merge)
//    verts.collect()
    val g2: Graph[(Int, Int), String] = Graph(verts, g.edges)

    //更新距离
    g2.vertices.collect()
    if(flag) {
      flag = false
//      println("if")
//      g2.vertices.collect().foreach(println(_))
      sumEdgeCount(g2)
    }

    else {
//      println("else")
//      g2.vertices.collect().foreach(println(_))
      g2
    }
  }

  def send(ec: EdgeContext[(Int, Int), String, (Int, Int)]): Unit = {

//    println("dest = " + ec.dstId + ",src = " + ec.srcId + ",send = " + (ec.srcAttr._1 + 1) + ",max = " + ec.dstAttr._2)
    if(ec.srcId == 1){
      ec.sendToSrc((0, 0))
    }
    ec.sendToDst((ec.srcAttr._1 + 1, ec.dstAttr._2))
  }

  def merge(a: (Int, Int), b: (Int, Int)): (Int, Int) = {
    val attr = math.max(a._1, b._1)
//    println("a2 = " + a._2 + ",b2 = " + b._2 + ",attr = " + attr)
    if(attr > a._2){
//      println("--------------if---------------")
      flag = true
      (attr, attr)
    }
    else {
//      println("--------------else---------------")
      (attr, a._2)
    }
  }

  def main(args: Array[String]): Unit = {

    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    // 构建图
    val myVertices = sc.parallelize(Array((1L, "张三"), (2L, "李四"), (3L, "王五"), (4L, "钱六"),
      (5L, "领导")))
    val myEdges = sc.makeRDD(Array(Edge(1L, 2L, "朋友"),
      Edge(2L, 3L, "朋友"), Edge(3L, 4L, "朋友"),
      Edge(4L, 5L, "上下级"), Edge(3L, 5L, "上下级")
    ))

    val myGraph = Graph(myVertices, myEdges)

    val initGraph: Graph[(Int, Int), String] = myGraph.mapVertices((id, eg) => {
      //第一个0是距离 第二个0是改顶点的最远距离
      (0, 0)
    })

//    initGraph.vertices.collect().foreach(println(_))
//    sumEdgeCount(initGraph).vertices.collect.foreach(println(_))
    sumEdgeCount(initGraph).vertices.collect().foreach(println(_))
//    sumEdgeCount(initGraph)
  }
}
